package com.qzxl.conntext.kafka;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Properties;
import java.util.regex.Pattern;

/**
 * kafka消费者/监听
 *
 * @auther fanzhengwei
 * @date 2019/12/26
 */
public class KafkaConsumerConfig1 implements Runnable {

    public KafkaConsumerConfig1() {
    }

    public KafkaConsumerConfig1(String topic, String servers, boolean enableAutoCommit, String sessionTimeout, String autoCommitInterval, String groupId, String autoOffsetReset) {
        this.topic = topic;
        this.servers = servers;
        this.enableAutoCommit = enableAutoCommit;
        this.sessionTimeout = sessionTimeout;
        this.autoCommitInterval = autoCommitInterval;
        this.groupId = groupId;
        this.autoOffsetReset = autoOffsetReset;
    }

    private String topic; //订阅topic =--key2{222} --key1,{111}
    private String servers; //kafka地址
    private boolean enableAutoCommit;//是否自动提交
    private String sessionTimeout;
    private String autoCommitInterval;
    private String groupId;
    private String autoOffsetReset;

    public KafkaConsumer<String, String> kafkaConsumer() {
        Properties properties = new Properties();
        properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, servers);
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, autoCommitInterval);
        properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeout);
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
        return new KafkaConsumer<String, String>(properties);
    }

    @Override
    public void run() {
        KafkaConsumer kafkaConsumer = kafkaConsumer();
        kafkaConsumer.subscribe(Pattern.compile(topic));
        while (true) {
            ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(100));
            if (!records.isEmpty()) {
                try {
                    for (ConsumerRecord<String, String> record : records) {
                        JSONObject json = JSON.parseObject(record.value());
                        System.out.println("*****************************************************");
                        System.out.println(json);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

}